package com.michale.admin.mqtt.config;

import com.michale.common.properties.MQTTProperties;
import com.michale.common.utils.uuid.IdUtils;
import io.swagger.annotations.Api;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent;
import org.springframework.integration.mqtt.event.MqttMessageSentEvent;
import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;
import java.security.SecureRandom;

import static com.michale.common.constants.MqttConstant.MQTT_OUTBOUND_CHANNEL;

/**
 * @author Michale
 * @apiNote MQTT配置类
 */
@Slf4j
@Api(tags = "MQTT配置类")
@Configuration
public class MqttConfig {
    @Resource
    private MQTTProperties MQTTProperties;

    /**
     * 创建MqttPahoClientFactory
     * 设置MQTT Broker连接属性
     *
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(MQTTProperties.getUsername());
        options.setPassword(MQTTProperties.getPassword().toCharArray());
        options.setKeepAliveInterval(MQTTProperties.getKeepalive());
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(MQTTProperties.getTimeout());
        options.setMaxInflight(1000000);
        //多个服务器地址时处理
        options.setServerURIs(MQTTProperties.getHostUrl().split(","));
        factory.setConnectionOptions(options);
        return factory;
    }

    /**
     * 发送通道配置 默认主题
     *
     * @return messageHandler
     */
    @Bean
    @ServiceActivator(inputChannel = MQTT_OUTBOUND_CHANNEL)
    public MessageHandler mqttOutbound() {
        //clientId每个连接必须唯一,否则,两个相同的clientId相互挤掉线
        String clientIdStr = MQTTProperties.getClientId() + new SecureRandom().nextInt(10);
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientIdStr, mqttClientFactory());
        //async如果为true，则调用方不会阻塞。而是在发送消息时等待传递确认。默认值为false（发送将阻塞，直到确认发送)
        messageHandler.setDefaultTopic(MQTTProperties.getDefaultTopic());
        messageHandler.setAsync(true);
        messageHandler.setAsyncEvents(true);
        messageHandler.setDefaultQos(0);
        return messageHandler;
    }


    /**
     * @apiNote  配置监听的 topic 支持通配符
     */
    @Bean
    public MessageProducer inbound() {
        //clientId每个连接必须唯一,否则,两个相同的clientId相互挤掉线
        String serverIdStr = MQTTProperties.getClientId() + IdUtils.fastUUID();
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(serverIdStr, mqttClientFactory(), MQTTProperties.getDefaultTopic());
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setCompletionTimeout(MQTTProperties.getTimeout());
        adapter.setQos(0);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }


    /**
     * @apiNote 当async和async事件(async - events)都为true时, 将发出MqttMessageSentEvent
     * 它包含消息、主题、客户端库生成的消息id、clientId和clientInstance（每次连接客户端时递增）
     */
    @EventListener(MqttMessageSentEvent.class)
    public void mqttMessageSentEvent(MqttMessageSentEvent event) {
        log.info("发送信息: info={}", event.toString());
    }

    /**
     * @apiNote  当async和async事件(async - events)都为true时, 将发出MqttMessageDeliveredEvent
     * 当客户端确认传递时，将发出MqttMessageDeliveredEvent
     * 它包含messageId、clientId和clientInstance，使传递与发送相关。
     */
    @EventListener(MqttMessageDeliveredEvent.class)
    public void mqttMessageDeliveredEvent(MqttMessageDeliveredEvent event) {
        log.info("发送成功信息:  info={}", event.toString());
    }

    /**
     * @apiNote 成功订阅到主题
     * @date 2021/7/22
     */
    @EventListener(MqttSubscribedEvent.class)
    public void mqttSubscribedEvent(MqttSubscribedEvent event) {

        log.info("成功订阅到主题:  info={}", event.toString());
    }

    /**
     * @apiNote  发送通道
     */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * @apiNote 接收通道
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
}
